package com.duoec.dw.lineage.service.impl;

import com.duoec.dw.lineage.dto.TableCoordinateDto;
import com.duoec.dw.lineage.dto.TableFieldDto;
import com.duoec.dw.lineage.dto.TableInfoDto;
import com.duoec.dw.lineage.service.LineageTableService;
import com.google.common.base.Charsets;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.io.Resources;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.io.StringReader;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;

/**
 * 仅为测试使用！！
 *
 * @author xuwenzhen
 */
public class LineageTableServiceHardCodeImpl implements LineageTableService {
    private static final Logger logger = LoggerFactory.getLogger(LineageTableServiceHardCodeImpl.class);

    protected static final Set<String> TABLES = Sets.newHashSet();
    public static final String EMPTY_STR = "";
    public static final String STR_COL_NAME = "col_name";
    public static final String STR_DATA_TYPE = "data_type";

    /**
     * 获取表字段信息
     *
     * @param tableCoordinate 数据表
     * @return 字段信息列表
     */
    @Override
    public TableInfoDto getTableDto(TableCoordinateDto tableCoordinate) {
        if (tableCoordinate.getTableName() == null) {
            return null;
        }
        String tableId = tableCoordinate.toString();

        //尝试从运行时中加载
        TableInfoDto table = LineageTableService.getTable(tableId);
        if (table != null) {
            return table;
        }

        String str;
        try {
            str = Resources.toString(Resources.getResource("table/" + tableCoordinate.getDatabaseName() + "/" + tableCoordinate.getTableName() + ".csv"), Charsets.UTF_8);
        } catch (Exception e) {
            logger.warn("需要加载数据表信息：{}", tableId);
            TABLES.add(tableId);
            return null;
        }

        TableInfoDto dto = new TableInfoDto();
        dto.setDatabaseName(tableCoordinate.getDatabaseName());
        dto.setTableName(tableCoordinate.getTableName());
        dto.setComment(EMPTY_STR);
        LinkedHashSet<TableFieldDto> fields = new LinkedHashSet<>();
        dto.setFields(fields);

        Iterable<CSVRecord> records;
        try {
            records = CSVFormat.EXCEL.withQuote('"').parse(new StringReader(str));
        } catch (IOException e) {
            logger.error("读取表信息失败！", e);
            return null;
        }
        Map<String, TableFieldDto> fieldMap = Maps.newHashMap();
        boolean partition = false;
        for (CSVRecord record : records) {
            String columnName = record.get(0);
            String columnType = record.get(1);
            if (CollectionUtils.isEmpty(fields) && STR_COL_NAME.equals(columnName) && STR_DATA_TYPE.equals(columnType)) {
                continue;
            }
            if (StringUtils.isEmpty(columnName)) {
                continue;
            }
            if (columnName.startsWith("#")) {
                if (columnName.startsWith("# Partition")) {
                    partition = true;
                }
                continue;
            }
            TableFieldDto field = fieldMap.get(columnName);
            if (field == null) {
                String columnComment = record.get(2);
                field = new TableFieldDto();
                field.setColName(columnName);
                field.setDataType(columnType);
                field.setComment(columnComment);
                field.setIndex(fields.size());
                fields.add(field);
                fieldMap.put(columnName, field);
            }
            field.setFieldType(partition ? 2 : 1);
        }

        return dto;
    }
}
